//! CDB查询优化器集成测试
//! 
//! 本文件包含对CDB查询优化器的全面测试，验证统计信息收集、成本估算、优化规则等功能

use cdb::sql::optimizer::*;
use cdb::sql::planner::{LogicalPlan, Expression, LiteralValue};
use cdb::sql::Value;
use std::sync::Arc;

/// 创建测试数据
fn create_test_data() -> Vec<Vec<Value>> {
    vec![
        vec![Value::Integer(1), Value::Text("Alice".to_string()), Value::Integer(25)],
        vec![Value::Integer(2), Value::Text("Bob".to_string()), Value::Integer(30)],
        vec![Value::Integer(3), Value::Text("Charlie".to_string()), Value::Integer(35)],
        vec![Value::Integer(4), Value::Text("Diana".to_string()), Value::Integer(28)],
        vec![Value::Integer(5), Value::Text("Eve".to_string()), Value::Integer(32)],
    ]
}

/// 测试查询优化器创建
#[tokio::test]
async fn test_query_optimizer_creation() {
    let optimizer = QueryOptimizer::new();
    let stats = optimizer.get_stats();
    
    assert_eq!(stats.optimization_count, 0);
    assert_eq!(stats.rule_applications, 0);
    assert_eq!(stats.avg_optimization_time_us, 0);
    
    println!("✅ 查询优化器创建测试通过");
}

/// 测试统计信息收集器
#[tokio::test]
async fn test_statistics_collector() {
    let collector = StatisticsCollector::new();
    let test_data = create_test_data();
    
    // 收集表统计信息
    let result = collector.collect_table_statistics("users", &test_data);
    assert!(result.is_ok());
    
    // 验证表统计信息
    let table_stats = collector.get_table_statistics("users");
    assert!(table_stats.is_some());
    
    let stats = table_stats.unwrap();
    assert_eq!(stats.table_name, "users");
    assert_eq!(stats.row_count, 5);
    assert!(stats.avg_row_size > 0.0);
    assert!(stats.page_count > 0);
    
    println!("📊 表统计信息:");
    println!("  表名: {}", stats.table_name);
    println!("  行数: {}", stats.row_count);
    println!("  平均行大小: {:.2} 字节", stats.avg_row_size);
    println!("  页面数: {}", stats.page_count);
    
    // 收集列统计信息
    let result = collector.collect_column_statistics("users", "name", 1, &test_data);
    assert!(result.is_ok());
    
    // 验证列统计信息
    let column_stats = collector.get_column_statistics("users", "name");
    assert!(column_stats.is_some());
    
    let col_stats = column_stats.unwrap();
    assert_eq!(col_stats.table_name, "users");
    assert_eq!(col_stats.column_name, "name");
    assert_eq!(col_stats.distinct_count, 5); // 5个不同的名字
    assert_eq!(col_stats.null_count, 0); // 没有NULL值
    
    println!("📊 列统计信息:");
    println!("  列名: {}.{}", col_stats.table_name, col_stats.column_name);
    println!("  唯一值数量: {}", col_stats.distinct_count);
    println!("  NULL值数量: {}", col_stats.null_count);
    
    println!("✅ 统计信息收集器测试通过");
}

/// 测试成本估算模型
#[tokio::test]
async fn test_cost_model() {
    let collector = Arc::new(StatisticsCollector::new());
    let test_data = create_test_data();
    
    // 收集统计信息
    collector.collect_table_statistics("users", &test_data).unwrap();
    
    let cost_model = CostModel::new(collector.clone());
    
    // 测试表扫描成本估算
    let table_scan = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let cost = cost_model.estimate_cost(&table_scan);
    assert!(cost.is_ok());
    
    let scan_cost = cost.unwrap();
    assert!(scan_cost > 0.0);
    
    println!("💰 成本估算结果:");
    println!("  表扫描成本: {:.4}", scan_cost);
    
    // 测试过滤操作成本
    let filter_plan = LogicalPlan::Filter {
        input: Box::new(table_scan.clone()),
        predicate: Expression::Literal(LiteralValue::Boolean(true)),
    };
    
    let filter_cost = cost_model.estimate_cost(&filter_plan);
    assert!(filter_cost.is_ok());
    
    let filter_cost_value = filter_cost.unwrap();
    assert!(filter_cost_value > scan_cost); // 过滤操作应该比单纯扫描成本更高
    
    println!("  过滤操作成本: {:.4}", filter_cost_value);
    
    // 测试投影操作成本
    let projection_plan = LogicalPlan::Projection {
        input: Box::new(table_scan),
        expressions: vec![Expression::Literal(LiteralValue::String("name".to_string()))],
    };
    
    let projection_cost = cost_model.estimate_cost(&projection_plan);
    assert!(projection_cost.is_ok());
    
    let proj_cost_value = projection_cost.unwrap();
    assert!(proj_cost_value > 0.0);
    
    println!("  投影操作成本: {:.4}", proj_cost_value);
    
    // 测试成本参数
    let params = cost_model.get_parameters();
    assert!(params.seq_scan_cost_per_row > 0.0);
    assert!(params.cpu_cost_per_row > 0.0);
    assert!(params.sequential_page_cost > 0.0);
    
    println!("💰 成本参数:");
    println!("  顺序扫描成本/行: {:.6}", params.seq_scan_cost_per_row);
    println!("  CPU成本/行: {:.6}", params.cpu_cost_per_row);
    println!("  页面成本: {:.6}", params.sequential_page_cost);
    
    println!("✅ 成本估算模型测试通过");
}

/// 测试计划缓存
#[tokio::test]
async fn test_plan_cache() {
    let cache = PlanCache::new();
    
    // 创建测试计划
    let plan = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let query_hash = "SELECT * FROM users".to_string();
    
    // 测试缓存存储
    cache.put(query_hash.clone(), plan.clone(), 100.0); // 添加成本参数

    // 测试缓存检索
    let cached_plan = cache.get(&query_hash);
    assert!(cached_plan.is_some());

    // 验证缓存统计信息
    let stats = cache.get_stats();
    assert_eq!(stats.current_entries, 1);
    assert_eq!(stats.hits, 1); // get操作算作一次命中
    
    // 测试缓存命中率
    let hit_rate = cache.hit_rate();
    assert!(hit_rate >= 0.0 && hit_rate <= 1.0);
    
    println!("🗄️ 计划缓存统计:");
    println!("  缓存条目数: {}", stats.current_entries);
    println!("  缓存命中数: {}", stats.hits);
    println!("  缓存未命中数: {}", stats.misses);
    println!("  命中率: {:.2}%", hit_rate * 100.0);

    // 测试缓存清理
    cache.clear();
    let stats_after_clear = cache.get_stats();
    assert_eq!(stats_after_clear.current_entries, 0);
    
    println!("✅ 计划缓存测试通过");
}

/// 测试优化规则
#[tokio::test]
async fn test_optimization_rules() {
    // 测试谓词下推规则
    let predicate_rule = PredicatePushdownRule;
    assert_eq!(predicate_rule.name(), "PredicatePushdown");
    
    let table_scan = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let filter_plan = LogicalPlan::Filter {
        input: Box::new(table_scan),
        predicate: Expression::Literal(LiteralValue::Boolean(true)),
    };
    
    let optimized = predicate_rule.optimize(filter_plan);
    assert!(optimized.is_ok());
    
    // 测试投影下推规则
    let projection_rule = ProjectionPushdownRule;
    assert_eq!(projection_rule.name(), "ProjectionPushdown");
    
    let table_scan2 = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let projection_plan = LogicalPlan::Projection {
        input: Box::new(table_scan2),
        expressions: vec![Expression::Literal(LiteralValue::String("name".to_string()))],
    };
    
    let optimized2 = projection_rule.optimize(projection_plan);
    assert!(optimized2.is_ok());
    
    // 测试常量折叠规则
    let constant_rule = ConstantFoldingRule;
    assert_eq!(constant_rule.name(), "ConstantFolding");
    
    // 测试JOIN重排序规则
    let join_rule = JoinReorderRule;
    assert_eq!(join_rule.name(), "JoinReorder");
    
    // 测试索引选择规则
    let index_rule = IndexSelectionRule;
    assert_eq!(index_rule.name(), "IndexSelection");
    
    println!("🔧 优化规则测试:");
    println!("  谓词下推规则: {}", predicate_rule.name());
    println!("  投影下推规则: {}", projection_rule.name());
    println!("  常量折叠规则: {}", constant_rule.name());
    println!("  JOIN重排序规则: {}", join_rule.name());
    println!("  索引选择规则: {}", index_rule.name());
    
    println!("✅ 优化规则测试通过");
}

/// 测试完整的查询优化流程
#[tokio::test]
async fn test_complete_optimization_flow() {
    let mut optimizer = QueryOptimizer::new();
    let test_data = create_test_data();
    
    // 创建统计信息收集器
    let collector = Arc::new(StatisticsCollector::new());
    collector.collect_table_statistics("users", &test_data).unwrap();
    
    // 设置统计信息收集器
    optimizer.set_statistics_collector(collector);
    
    // 创建复杂的查询计划
    let table_scan = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let filter_plan = LogicalPlan::Filter {
        input: Box::new(table_scan),
        predicate: Expression::Literal(LiteralValue::Boolean(true)),
    };
    
    let projection_plan = LogicalPlan::Projection {
        input: Box::new(filter_plan),
        expressions: vec![
            Expression::Literal(LiteralValue::String("id".to_string())),
            Expression::Literal(LiteralValue::String("name".to_string())),
        ],
    };
    
    // 执行优化
    let start_time = std::time::Instant::now();
    let optimized_plan = optimizer.optimize(projection_plan);
    let optimization_time = start_time.elapsed();
    
    assert!(optimized_plan.is_ok());
    
    // 验证优化统计信息
    let stats = optimizer.get_stats();
    assert_eq!(stats.optimization_count, 1);
    assert!(stats.rule_applications > 0);
    assert!(stats.avg_optimization_time_us > 0);
    
    println!("🚀 完整优化流程测试:");
    println!("  优化次数: {}", stats.optimization_count);
    println!("  规则应用次数: {}", stats.rule_applications);
    println!("  平均优化时间: {} μs", stats.avg_optimization_time_us);
    println!("  本次优化时间: {:?}", optimization_time);
    
    println!("✅ 完整优化流程测试通过");
}

/// 测试优化器性能
#[tokio::test]
async fn test_optimizer_performance() {
    let mut optimizer = QueryOptimizer::new();
    let test_data = create_test_data();
    
    // 创建统计信息收集器
    let collector = Arc::new(StatisticsCollector::new());
    collector.collect_table_statistics("users", &test_data).unwrap();
    optimizer.set_statistics_collector(collector);
    
    // 创建测试计划
    let table_scan = LogicalPlan::TableScan {
        table_name: "users".to_string(),
        alias: None,
        projection: None,
    };
    
    let num_optimizations = 100;
    let start_time = std::time::Instant::now();
    
    // 执行多次优化
    for _ in 0..num_optimizations {
        let result = optimizer.optimize(table_scan.clone());
        assert!(result.is_ok());
    }
    
    let total_time = start_time.elapsed();
    let avg_time_per_optimization = total_time / num_optimizations;
    
    // 验证性能要求
    assert!(avg_time_per_optimization.as_micros() < 1000); // 每次优化应该小于1ms
    
    let stats = optimizer.get_stats();
    assert_eq!(stats.optimization_count, num_optimizations as u64);
    
    println!("⚡ 优化器性能测试:");
    println!("  优化次数: {}", num_optimizations);
    println!("  总时间: {:?}", total_time);
    println!("  平均时间/次: {:?}", avg_time_per_optimization);
    println!("  吞吐量: {:.2} 优化/秒", num_optimizations as f64 / total_time.as_secs_f64());
    
    // 性能要求：至少1000次优化/秒
    let throughput = num_optimizations as f64 / total_time.as_secs_f64();
    assert!(throughput >= 1000.0, "优化器吞吐量过低: {:.2} 优化/秒", throughput);
    
    println!("✅ 优化器性能测试通过");
}

/// 测试统计信息配置
#[tokio::test]
async fn test_statistics_config() {
    let config = StatisticsConfig {
        max_age_seconds: 3600,
        histogram_buckets: 100,
        auto_update_threshold: 0.1,
        sampling_rate: 0.01,
    };

    let collector = StatisticsCollector::with_config(config.clone());

    // 验证配置（通过创建测试数据来间接验证配置生效）
    let test_data = create_test_data();
    let result = collector.collect_table_statistics("test_table", &test_data);
    assert!(result.is_ok());

    // 验证统计信息收集成功
    let table_stats = collector.get_table_statistics("test_table");
    assert!(table_stats.is_some());

    println!("⚙️ 统计信息配置:");
    println!("  最大存活时间: {} 秒", config.max_age_seconds);
    println!("  直方图桶数: {}", config.histogram_buckets);
    println!("  自动更新阈值: {:.1}%", config.auto_update_threshold * 100.0);
    println!("  采样率: {:.1}%", config.sampling_rate * 100.0);

    println!("✅ 统计信息配置测试通过");
}
